package com.service;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mk.HibernateException;
import org.mk.type.IntegerType;
import org.mk.type.MySQLTypeFactory;
import org.mk.type.Type;

import com.client.RecordCreateService;
import com.db.ColumnInfoBean;
import com.db.ForeignKey;
import com.db.connection.DBCPPostgresManager;
import com.db.domain.Communication;
import com.db.domain.Record;
import com.db.sql.convert.DatabaseConverter;
import com.db.sql.err.ForeignErrorStragety;
import com.excel.read.Config;
import com.excel.read.DBTable;
import com.operation.OperationMap;
import com.operation.POXML;
import com.operation.RecordOperation;
import com.operation.SQLOperationMap;
import com.operation.TreeTool;
import com.operation.model.OperationModel;
import com.xml.ExtractFactory;
import com.xml.RecordExtract;
import com.xml.SQLRecordExtract;

/**
 * 
 * @功能 将记录导到数据库中 单线程处理
 * @author 张华
 * @date May 23, 2009
 * @version 1.0 May 23, 2009 12:01:19 PM
 */
public class ImportService implements Runnable {
	public static Log log = LogFactory.getLog(ImportService.class);
	public static final int PO_FLAG = 1;
	public static final int SQL_FLAG = 2;

	public static final int INSERT = 1;
	public static final int UPDATE = 2;
	public static final int DELETE = 3;

	public String getRecordText(OperationMap a, com.xml.Record record,
			Connection conn) throws HibernateException, SQLException {
		DBTable dbTable = (DBTable) Config.allTable.get(record.getTableName());
		List<ColumnInfoBean> cinfoNameList = dbTable.getColumnInfos();
		List<String> selfNameList = new ArrayList<String>();
		for (ColumnInfoBean cinfo : cinfoNameList) {
			selfNameList.add(cinfo.getName());
		}
		String sql = dbTable.getSelectSQL();
		ColumnInfoBean pk = dbTable.getPk();
		// 执行SQL语句
		PreparedStatement pstmt = conn.prepareStatement(sql);
		Type atype = MySQLTypeFactory.basic(pk.getNoPresitionTypeName());
		Object value = record.fieldMap.get(pk.getName().toUpperCase());
		atype.nullSafeSet(pstmt, value, 1);
		ResultSet rs = pstmt.executeQuery();
		String xml = "";
		if (rs.next()) {
			xml = POXML.getXML(dbTable, rs);
		}

		return xml;
	}

	/**
	 * 取得请求并且处理
	 * 
	 * @param requestList
	 * @throws Exception
	 */
	public void doResponse(List<Communication> requestList, Connection conn)
			throws Exception {
		for (Communication aCommunication : requestList) {
			RecordExtract aRecordExtract = new RecordExtract();
			OperationModel aOperationMap = aRecordExtract.extractOperationMap(
					aCommunication.request, aCommunication.clientFrom);
			List<OperationModel> list = new ArrayList<OperationModel>();
			list.add(aOperationMap);
			String response = ((OperationMap) aOperationMap)
					.getRecordText(conn);
			aCommunication.setResponse(response);
		}

	}

	/**
	 * 判断Record的状态，如果为 N，表示未访问过该记录 Y:访问过，无需外键 B：访问过需要外键 取得Record 1.0
	 * 查看与该record相关联的最新Request，如果有加载Request，如果没有不加载Request， 1.0 根据record ， 1.1
	 * 取得外键， 查看是否在本地存在，如果存在，继续，如果不存在
	 * 
	 * 只有当整棵树加载完毕才处理Record
	 * 
	 * @param rd
	 *            待操作的记录
	 * @param conn
	 * @return
	 * @throws Exception
	 * @date May 23, 2009
	 */
	public static boolean operationPO(Record rd, Connection conn)
			throws Exception {
		CCService aCCService = SyncServiceFactory.getCCService();
		List<Communication> reqList = new ArrayList<Communication>();
		RecordExtract aRecordExtract = ExtractFactory.recordExtract;
		// PO
		// 待操作的记录
		OperationModel aOperationMap2 = aRecordExtract.extractOperationMap(
				rd.content, rd.storeFrom);
		List<OperationModel> list = new ArrayList<OperationModel>();
		list.add(aOperationMap2);

		OperationMap aOperationMap = (OperationMap) aOperationMap2;
		// 第一次访问，需要判断状态
		if (rd.fkFlag == null || rd.fkFlag.equalsIgnoreCase("N")
				|| rd.fkFlag.trim().equals("")) {
			List<Communication> mList = null;
			// 判断根记录是否需要外键请求
			List<Communication> clist = aOperationMap.getRequestInit(null);
			// 以后的处理由Communication完成，不再走Record, 但是Record的状态以后设置为N
			// 直到Communication树操作完成
			if (clist.size() > 0) {
				// 添加请求，新增Commnuication，该Commnuication为根
				Communication aCommunication = new Communication();
				aCommunication.request = "";
				aCommunication.response = rd.content;
				aCommunication.clientFrom = RecordCreateService.orgid;
				// 来自哪里，就发到哪里
				aCommunication.clientTo = rd.storeFrom;
				aCommunication.flag = "Y";
				aCommunication.parent = 0;
				aCommunication.level = 1;
				aCommunication.created = "";
				aCommunication.recordId = rd.id;
				aCCService.insert(aCommunication);
				setFKFlag(rd.id, "B");
			} else {
				// 处理
				RecordOperation.insert(list, conn);
				setFKFlag(rd.id, "Y");
				return true;

			}

		}

		// 记录操作结束
		if (rd.fkFlag != null && rd.fkFlag.equalsIgnoreCase("Y")
				&& rd.flag != null && rd.flag.equalsIgnoreCase("Y")) {
			return true;
		}

		// 取得最record的请求响应
		List<Communication> aList = aCCService.client2Client(
				(int) RecordCreateService.orgid, (int) rd.id);
		// 响应在Record之前得到处理
		TreeTool aTreeTool = new TreeTool();
		Communication root = aTreeTool.getInitTree(aList);
//		Map nameIDTree = aTreeTool.getNameID(aList);		
		reqList = root.readyAll();		
		if (reqList.size() > 0) {
			// 发送请求
			aCCService.sendRequest(reqList);
		} else {
			if (aTreeTool.ready()) {
				// 操作所有，从叶子节点开始操作，属于后向排序
				doTree(aTreeTool.root, conn, (int) rd.storeFrom);
				return true;				
			}

		}

		return false;

	}

	public static boolean operationSQL(Record rd, Connection conn)
			throws Exception {
		CCService aCCService = SyncServiceFactory.getCCService();
		List<Communication> reqList = new ArrayList<Communication>();
		SQLRecordExtract aSQLRecordExtract = ExtractFactory.sQLRecordExtract;
		// PO
		// 待操作的记录
		SQLOperationMap aOperationMap = (SQLOperationMap) aSQLRecordExtract
				.extractOperationMap(rd.content, rd.storeFrom);

		String sql = aOperationMap.getSql();
		// 临时解决方案，对于insert()select 句子直接处理，不用解析，留在下一版本解析
		boolean tempFlag = false;

		/**
		 * 选择插入
		 */
		if ((sql.indexOf("insert") == 0 || sql.indexOf("INSERT") == 0)
				&& (sql.indexOf("select") > 0 || sql.indexOf("SELECT") > 0)) {
			tempFlag = true;
		}

		if (tempFlag || sql.indexOf("update") == 0
				|| sql.indexOf("UPDATE") == 0 || sql.indexOf("DELETE") == 0
				|| sql.indexOf("delete") == 0) {
			aOperationMap.sql = DatabaseConverter.getPostgresDatabase()
					.convertStatement(aOperationMap.sql);
			// 走外键Communication
			if (rd.fkFlag != null && rd.fkFlag.equalsIgnoreCase("B")) {
				return doTree(rd, conn);
			} else {
				aOperationMap.insert(conn);
				return true;
			}

		}

		// 单条插入,单条插入走的是PO,其实不应该走PO,应该走SQL，它的外键走PO
		aOperationMap.sql = aOperationMap.sql.replaceAll("SysDate",
				"TO_DATE('2010-08-05 09:29:53','YYYY-MM-DD HH24:MI:SS')");
		OperationMap operationMap = aOperationMap.convertToOperationMap();
		List<OperationMap> list = new ArrayList<OperationMap>();
		list.add(operationMap);

		// 第一次访问，需要判断状态
		if (rd.fkFlag == null || rd.fkFlag.equalsIgnoreCase("N")
				|| rd.fkFlag.equalsIgnoreCase("")) {
			List<Communication> mList = null;
			// 判断根记录是否需要外键请求
			List<Communication> clist = operationMap.getRequest(null);
			// 以后的处理由Communication完成，不再走Record, 但是Record的状态以后设置为N
			// 直到Communication树操作完成
			if (clist.size() > 0) {
				// 添加请求，新增Commnuication，该Commnuication为根
				Communication aCommunication = new Communication();
				aCommunication.request = "";
				aCommunication.response = rd.content;
				aCommunication.clientFrom = RecordCreateService.orgid;
				// 来自哪里，就发到哪里
				aCommunication.clientTo = rd.storeFrom;
				aCommunication.flag = "Y";
				aCommunication.parent = 0;
				aCommunication.level = 1;
				aCommunication.created = "";
				aCommunication.recordId = rd.id;
				aCCService.insert(aCommunication);
				setFKFlag(rd.id, "B");

			} else {
				// 处理
				aOperationMap.insert(conn);
				setFKFlag(rd.id, "Y");
				return true;

			}

		}
		RecordExtract aRecordExtract = ExtractFactory.recordExtract;
		// 记录操作结束
		if (rd.fkFlag != null && rd.fkFlag.equalsIgnoreCase("Y")
				&& rd.flag != null && rd.flag.equalsIgnoreCase("Y")) {
			return true;
		}

		// 取得最record的请求响应
		List<Communication> aList = aCCService.client2Client(
				(int) RecordCreateService.orgid, (int) rd.id);
		// 响应在Record之前得到处理
		TreeTool aTreeTool = new TreeTool();
		Communication root = aTreeTool.getInitTree(aList);
//		Map nameIDTree = aTreeTool.getNameID(aList);		
		reqList = root.readyAll();		
		if (reqList.size() > 0) {
			// 发送请求
			aCCService.sendRequest(reqList);
		} else {
			if (aTreeTool.ready()) {
				// 操作所有，从叶子节点开始操作，属于后向排序
				doTree(aTreeTool.root, conn, (int) rd.storeFrom);
				return true;
				// RecordOperation.insert(list, conn);
			}

		}

		return false;

	}

	/**
	 * 取得Communication树并处理 所有走Communication的都是PO
	 * 
	 * @param rd
	 * @param toTableName
	 * @param toFieldname
	 * @param value
	 *            Aug 8, 2009
	 * @throws Exception
	 */
	public static boolean doTree(Record rd, Connection conn) throws Exception {
		// 记录操作结束
		if (rd.fkFlag != null && rd.fkFlag.equalsIgnoreCase("Y")
				&& rd.flag != null && rd.flag.equalsIgnoreCase("Y")) {
			return true;
		}

		List<Communication> reqList = new ArrayList<Communication>();
		CCService aCCService = SyncServiceFactory.getCCService();
		// 取得最record的请求响应
		List<Communication> aList = aCCService.client2Client(
				(int) RecordCreateService.orgid, (int) rd.id);
		
		//取主键和表名，防止树的死循环
		// 响应在Record之前得到处理
		TreeTool aTreeTool = new TreeTool();
		
		//Map nameIdMap = aTreeTool.getNameID(aList);
		aTreeTool.getTree(aList);
		List<Communication> leafList = aTreeTool.leafList;
		RecordExtract aRecordExtract = new RecordExtract();
		for (Communication aCommunication : leafList) {
			// 处理每一个响应,响应可能是SQL、PO
			if (aCommunication.ready()) {
				int opType = PO_FLAG;
				if (!ExtractFactory.isPO(aCommunication.response)) {
					opType = SQL_FLAG;
				}
				OperationMap aOperationMap2 = null;
				if (opType == PO_FLAG) {
					aOperationMap2 = (OperationMap) aRecordExtract
							.extractOperationMap(aCommunication.response,
									rd.storeFrom);
				}

				List<Communication> clist = aOperationMap2
						.getRequest(aCommunication);
				reqList.addAll(clist);
			}

		}
		if (reqList.size() > 0) {
			// 发送请求
			aCCService.sendRequest(reqList);
		} else {
			if (aTreeTool.ready()) {
				// 操作所有，从叶子节点开始操作，属于后向排序
				doTree(aTreeTool.root, conn, (int) rd.storeFrom);
				return true;
				// RecordOperation.insert(list, conn);
			}

		}

		return false;
	}

	/**
	 * 根据Record直接生成Communication,用于根记录
	 * 
	 * @param rd
	 * @param toTableName
	 * @param toFieldname
	 * @param value
	 *            Aug 15, 2009
	 */
	public static void createRootCommunicationByException(Record rd,
			String toTableName, String toFieldname, String value) {
		CCService aCCService = SyncServiceFactory.getCCService();
		Integer rootId = aCCService.recordExist(
				(int) RecordCreateService.orgid, rd.id);
		if (rootId == null) {
			Communication aCommunication = new Communication();
			aCommunication.request = "";
			aCommunication.response = rd.content;
			aCommunication.clientFrom = RecordCreateService.orgid;
			// 来自哪里，就发到哪里
			aCommunication.clientTo = rd.storeFrom;
			aCommunication.flag = "Y";
			aCommunication.parent = 0;
			aCommunication.level = 1;
			aCommunication.created = "";
			aCommunication.recordId = rd.id;
			aCCService.insert(aCommunication);
			rootId = aCCService.recordExist((int) RecordCreateService.orgid,
					rd.id);
		} else {

		}

		Communication aCommunication2 = new Communication();
		String xml = POXML.getRecord(toTableName, toFieldname, Long
				.parseLong(value));
		aCommunication2.request = xml;
		aCommunication2.clientFrom = RecordCreateService.orgid;
		// 来自哪里，就发到哪里
		aCommunication2.clientTo = rd.storeFrom;
		aCommunication2.flag = "N";
		aCommunication2.parent = rootId;
		aCommunication2.level = 2;
		aCommunication2.recordId = rd.id;
		aCCService.insert(aCommunication2);
		setFKFlag(rd.id, "B");
	}

	public static boolean operationSQLUPdateDelete(Record rd, Connection conn)
			throws Exception {
		// SQL
		SQLRecordExtract aSQLRecordExtract = new SQLRecordExtract();
		SQLOperationMap aOperationMap = aSQLRecordExtract.extractOperationMap(
				rd.content, rd.storeFrom);
		aOperationMap.insert(conn);
		return true;
	}

	public static boolean operationSQLUPdateDelete2(
			SQLOperationMap aOperationMap, Connection conn) throws Exception {
		aOperationMap.insert(conn);
		return true;
	}

	public static boolean operationInsertSQL(Record rd, Connection conn)
			throws Exception {
		// SQL
		SQLRecordExtract aSQLRecordExtract = new SQLRecordExtract();
		SQLOperationMap aOperationMap = aSQLRecordExtract.extractOperationMap(
				rd.content, rd.storeFrom);
		aOperationMap.insert(conn);
		return true;
	}

	public static List<Record> get() {
		List<Record> aList = new ArrayList<Record>();
		Connection conn = null;
		PreparedStatement pstmt = null;
		ResultSet rs = null;
		ResultSet res = null;
		try {
			conn = DBCPPostgresManager.getConnection();
			// 执行SQL语句
			pstmt = conn
					.prepareStatement("select * from ad_record where  flag = 'N' order by id");
			Type type = new IntegerType();
			rs = pstmt.executeQuery();
			// while (rs.next()) {
			if (rs.next()) {
				Record aRecord = new Record();
				// aRecord.setTableName(tableName);
				long storeTo = rs.getLong("store_to");
				String operation = rs.getString("operation");
				String content = rs.getString("content");
				aRecord.content = content;
				aRecord.setOperation(operation);
				aRecord.storeId = storeTo;
				aRecord.id = rs.getLong("id");
				aRecord.storeFrom = rs.getLong("store_from");
				aRecord.fkFlag = rs.getString("fkflag");
				aList.add(aRecord);
			}
		} catch (Exception ex) {
			log.error("ErrorImport : " + ex.toString());
		} finally {
			DBCPPostgresManager.close(conn, pstmt, rs);

		}
		return aList;
	}

	public static boolean setFlag(List<Long> list, Connection conn)
			throws SQLException {
		PreparedStatement pstmt = null;
		ResultSet rs = null;
		ResultSet res = null;
		try {
			// 执行SQL语句
			String sql = "update ad_record set flag = 'Y' where id = ?";
			pstmt = conn.prepareStatement(sql);

			for (Long id : list) {
				pstmt.setLong(1, id);
				pstmt.addBatch();
			}
			pstmt.executeBatch();

		} finally {
			DBCPPostgresManager.close(null, pstmt, rs);

		}

		return true;
	}

	public void run() {
		List<Long> aList = null;
		while (true) {
			aList = null;
			aList = new ArrayList<Long>();
			// 去记录
			List<Record> list = get();
			// 插入数据
			// 事务开始
			Connection conn = null;
			Record rd = null;
			try {
				// 一次只取一条，因为前一条记录操作未完毕，后一条操作绝对不允许执行
				if (list.size() > 0) {
					conn = DBCPPostgresManager.getConnection();
					// 判断content是否是原始SQL
					rd = list.get(0);
					String type = "";

					boolean done = false;
					int opType = PO_FLAG;
					if (!ExtractFactory.isPO(rd.content)) {
						opType = SQL_FLAG;
					}

					if (opType == PO_FLAG) {
						done = operationPO(rd, conn);
					} else {
						done = operationSQL(rd, conn);
					}

					// 标记
					if (done) {
						aList.add(rd.getId());
						setFlag(aList, conn);
						conn.commit();
						if (aList.size() > 0) {
							log.info("导入结束:" + rd.getId());
						}
					}

				}

			} catch (SQLException e1) {
				try {
					conn.rollback();
				} catch (SQLException e) {
					System.out.print(e.toString());
				}
				String str = e1.toString();
				int aIndex = str.indexOf("violates unique constraint");
				int aIndex2 = str.indexOf("foreign key constraint");
				if (aIndex > 0) {
					str = str.substring(aIndex);
					setOneFlag(rd.id);
				} else if (aIndex2 > 0) {
					ForeignErrorStragety aForeignErrorStragety = new ForeignErrorStragety();
					aForeignErrorStragety.doError(str);
					//有可能网络中断
					try {
						createRootCommunicationByException(rd,
								aForeignErrorStragety.getTableName(),
								aForeignErrorStragety.getFieldName(),
								aForeignErrorStragety.getValue());
					} catch (Exception e2) {
						e2.printStackTrace();
					}
				}

				log.error(str);
			} catch (Exception e) {
				try {
					conn.rollback();
				} catch (SQLException eq) {
					e.printStackTrace();
				}
				e.printStackTrace();
			} finally {
				DBCPPostgresManager.close(conn, null, null);
			}

			try {
				Thread.currentThread().sleep(5000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

	}

	public static boolean setFKFlag(Long id, String flag) {
		Connection conn = null;
		PreparedStatement pstmt = null;
		ResultSet rs = null;
		try {

			conn = DBCPPostgresManager.getConnection();

			String sql = "update  ad_record  set  fkFlag = '" + flag
					+ "' where id = ?";
			pstmt = conn.prepareStatement(sql);
			pstmt.setLong(1, id);

			pstmt.executeUpdate();
			pstmt.getConnection().commit();

		} catch (SQLException ex) {
			log.error("ErrorServer : " + ex.toString());
		} finally {
			DBCPPostgresManager.close(conn, pstmt, rs);

		}

		return true;
	}

	public static boolean setOneFlag(long id) {
		Connection conn = null;
		PreparedStatement pstmt = null;
		ResultSet rs = null;
		ResultSet res = null;
		try {
			conn = DBCPPostgresManager.getConnection();
			// 执行SQL语句
			String sql = "update ad_record set flag = 'Y' where id = ?";
			pstmt = conn.prepareStatement(sql);
			pstmt.setLong(1, id);
			pstmt.executeUpdate();
			pstmt.getConnection().commit();

		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			DBCPPostgresManager.close(conn, pstmt, rs);

		}

		return true;
	}

	private static void doTree(Communication aCommunication, Connection conn,
			int storeFrom) throws Exception {
		List<Communication> list = new ArrayList<Communication>();
		RecordExtract aRecordExtract = new RecordExtract();
		if (aCommunication.hasChild()) {
			for (Communication bCommunication : aCommunication.children) {
				doTree(bCommunication, conn, storeFrom);
			}

			List<OperationModel> list2 = new ArrayList<OperationModel>();

			int opType = PO_FLAG;
			if (!ExtractFactory.isPO(aCommunication.response)) {
				opType = SQL_FLAG;
			}
			OperationMap aOperationMap2 = null;
			SQLRecordExtract aSQLRecordExtract = new SQLRecordExtract();
			if (opType == PO_FLAG) {
				aOperationMap2 = (OperationMap) aRecordExtract
						.extractOperationMap(aCommunication.response, storeFrom);

				list2.add(aOperationMap2);
				RecordOperation.insert(list2, conn);

			} else {
				SQLOperationMap operationMap2 = aSQLRecordExtract
						.extractOperationMap(aCommunication.response, storeFrom);

				if (operationMap2.sql.indexOf("INSERT") == 0
						&& operationMap2.sql.indexOf("SELECT") > 0) {
					operationMap2.sql = DatabaseConverter.getPostgresDatabase()
							.convertStatement(operationMap2.sql);
					operationMap2.insert(conn);
				} else {
					aOperationMap2 = operationMap2.convertToOperationMap();
					list2.add(aOperationMap2);
					RecordOperation.insert(list2, conn);
				}
			}

		} else {
			List<OperationModel> list2 = new ArrayList<OperationModel>();
			OperationMap aOperationMap2 = (OperationMap) aRecordExtract
					.extractOperationMap(aCommunication.response, storeFrom);
			list2.add(aOperationMap2);
			RecordOperation.insert(list2, conn);
		}
	}
	
	
	
	
	public static void main(String args[]) {
		
		
		
		
		
		
		
		
		
	}

}
